package cn.doitedu.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/2
 * @Desc: 学大数据，上多易教育
 *   从kafka中读取用户行为
 *   过滤出 所有的 ref=/x 的页面浏览事件
 *   将结果写入kafka
 **/
public class _01_Demo_1_SQL {
    public static void main(String[] args) {

        // api编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 包装成sql编程环境
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 在表环境中执行sql
        tenv.executeSql(
                "create table action_log(\n" +
                "   uid bigint,\n" +
                "   event_id string,\n" +
                "   properties map<string,string>,\n" +
                "   action_time bigint   \n" +
                ") with (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'doit46',\n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'g002',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'value.format' = 'json',\n" +
                "  'value.fields-include' = 'EXCEPT_KEY'\n" +
                ")");

        tenv.executeSql(
                "create table sink_table(\n" +
                "   uid bigint,\n" +
                "   event_id string,\n" +
                "   properties map<string,string>,\n" +
                "   action_time bigint   \n" +
                ") with (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'page-load-events',\n" +
                "  'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "  'properties.group.id' = 'g002',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'value.format' = 'json',\n" +
                "  'value.fields-include' = 'EXCEPT_KEY'\n" +
                ")");


        tenv.executeSql(
                "INSERT INTO  sink_table\n" +
                "SELECT         \n" +
                "    uid,       \n" +
                "    event_id,  \n" +
                "    properties,\n" +
                "    action_time\n" +
                "from action_log\n" +
                "where event_id = 'page_load'\n" +
                "and properties['ref']='/x'");



    }

}
